package com.example.springweb.rest;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.log4j.Log4j2;

/**
 * 基于spring boot的轻量级的实时通信协议
 * SSE 的优势：
 * 1.简单易用： SSE 基于 HTTP 协议，无需额外的协议或库，使用简单。
 * 2.单向通信： 适用于服务器向客户端推送数据的场景，例如消息提醒、实时数据更新等。
 * 3.自动重连： 当连接断开时，SSE 会自动尝试重新连接，保证通信的可靠性。
 */
@Log4j2
@Tag(name = "SSE实时通信的推送与订阅接口")
@RestController
@RequestMapping(value = "/sse")
public class SseController {

	private static final Map<String, SseEmitter> EMITTER_MAP = new ConcurrentHashMap<String, SseEmitter>();
	
	/**
	 * 客户端通过该接口订阅消息，返回一个 SseEmitter 对象，用于服务器端推送消息。
	 * @param id
	 * @return 接口会持续等待
	 */
	@Schema(title = "消息订阅接口")
	@GetMapping(path = "/subscribe/{id}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	public SseEmitter subscribe(@PathVariable String id) throws IOException {
		SseEmitter emitter = new SseEmitter(10 * 60 * 1000L);
		EMITTER_MAP.put(id, emitter);
		
		 // 模拟发送消息
        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    emitter.send(SseEmitter.event().data(Map.of("id", id, "message", "Message" + i)));
                    Thread.sleep(1000);
                }
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            }
//            emitter.complete();
        }).start();
		
		log.info("客户端订阅的ID: {}", id);
		
		return emitter;
	}

	/**
	 * 推送消息接口：推送消息的时候，subscribe 接口就会收到响应
	 * @param id
	 * @param message
	 */
	@Async
	@Schema(title = "消息推送接口")
	@PostMapping("/push/{id}")
	public void push(@PathVariable String id, @RequestBody String message) {
		SseEmitter emitter = EMITTER_MAP.get(id);
		if(emitter != null) {
			try {
				emitter.send(SseEmitter.event().data(Map.of("id", id, "message", message)));
				log.info("消息成功发送到客户端 ID: {}", id);
			} catch (Exception e) {
				// 异常删除缓存
				emitter.complete();
				EMITTER_MAP.remove(id);
				log.error("发送到客户端的消息失败: {}", id, e);
			}
		}
	}
	
	/**
	 * 关闭连接
	 * @param id
	 * @return
	 */
	@DeleteMapping("/disconnect/{id}")
	public String disconnect(@PathVariable String id) {
		 SseEmitter emitter = EMITTER_MAP.remove(id);
		 if(emitter != null) {
			 emitter.complete();
		 }
		 return "连接已关闭";
	}
}
